package com.example.producer;

import com.example.common.models.OrderLog;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by zhengwj on 2020/11/17 9:46.
 */
@Component
@Slf4j
public class LogProducer implements ApplicationRunner {
    @Resource
    private RocketMQTemplate rocketMQTemplate;
    @Value("${custom.message.size:10000}")
    private int messageSize;
    @Value("${custom.message.topic}")
    private String messageTopic;

    private AtomicInteger count = new AtomicInteger();

    public void run(ApplicationArguments args) throws Exception {
        buildSampleMessages().parallelStream().forEach(message -> rocketMQTemplate.asyncSend(messageTopic, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info(sendResult.toString());
                //log.info("计数器：{}",count.incrementAndGet());
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("消息发送异常", throwable);
            }
        }));
    }

    private List<Message<OrderLog>> buildSampleMessages() {
        List<Message<OrderLog>> datas = new ArrayList<>();
        for (int i = 1; i <= messageSize; i++) {
            OrderLog orderLog = new OrderLog();
            orderLog.setCustomerId(i);
            orderLog.setProductList(Arrays.asList((long) i, (long) messageSize + i));
            orderLog.setMoney(8888);
            orderLog.setAddress("收货地址" + i);
            orderLog.setOrderTime(System.currentTimeMillis());
            datas.add(MessageBuilder.withPayload(orderLog).build());
        }

        return datas;
    }
}
